package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;


/**
 * @Class:
 * @Author: xyl
 * @Description: 针对企查查 t_eci_assistance_judicial_partners_change 表和中数 t_judicial_ass_alter 表进行数据映射开发
 * @Date: 2021/12/22 15:30
 */
public class TEciAssistanceJudicialPartnersChange {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_eci_assistance_judicial_partners_change").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_judicial_ass_alter");
        long zs_count = zs_data.count();
        if(zs_count>0){
            /**
             * t_eci_assistance_equity_freeze 司法协助_股权变更
             */

            //获取中数删除数据表中的数据 --- 中数删除逻辑：公司的任何一条数据发生变化，先删除这个公司相关的该表的所有数据
            Dataset<Row> delete_data = spark.sql("select a.id,a.key_no,a.company_id,a.company_name,a.execution_matters,a.execution_verdict_num,a.executed_person_doc_type,a.executed_person_doc_num,\n" +
                    "a.assignee,a.assist_exec_date,a.assignee_doc_kind,a.assignee_reg_no,a.stock_company_name,\n" +
                    "c.jobid,'-1' as isadd \n" +
                    "from (select * from a_dc_ep_dwi.t_eci_assistance_judicial_partners_change where company_id is not null and company_id !='')a \n" +
                    "inner join a_dc_ep_ods.t_eci_company b \n" +
                    "on a.company_id=b.company_id \n" +
                    "inner join (select * from a_dc_ep_ods.zs_t_judicial_ass_del where creditcode is not null and creditcode !='')c \n" +
                    "on upper(trim(b.credit_code)) = upper(trim(c.creditcode)) \n");
            delete_data.createOrReplaceTempView("tmp_t_eci_assistance_judicial_partners_change");
            spark.sql("insert overwrite table a_dc_ep_incr.t_eci_assistance_judicial_partners_change select * from tmp_t_eci_assistance_judicial_partners_change");

            //获取中数增量数据表中的更新数据 （inner join）
            Dataset<Row> update_data = spark.sql("SELECT " +
                    "       d.RECORD_ID as RECORD_ID,\n" +
                    "       c.key_no,\n" +
                    "       c.company_id,\n" +
                    "       c.company_name,\n" +
                    "       a.EXECUTION,\n" +
                    "       a.JUDGMENT_NO,\n" +
                    "       case when a.INAME_LTYPE is not null then a.INAME_LTYPE else a.INAME_ITYPE end, \n" +
                    "       a.INAME_LICENCE,\n" +
                    "       a.ASSIGNEE,\n" +
                    "       a.AEDATE,\n" +
                    "       case when a.ASSIGNEE_LTYPE is not null then a.ASSIGNEE_LTYPE else a.ASSIGNEE_ITYPE end,\n" +
                    "       a.ASSIGNEE_LICENCE,\n" +
                    "       \"\",\n" +
                    "       a.JOBID,\n" +
                    "       \"0\" as isadd  " +
                    "FROM \n" +
                    "  (SELECT * FROM a_dc_ep_ods.zs_t_judicial_ass_alter WHERE ENTID IS NOT NULL AND ENTID !='') a \n" +
                    "INNER JOIN \n" +
                    "  (SELECT * FROM a_dc_ep_ods.zs_t_judicial_ass_del WHERE creditcode IS NOT NULL AND creditcode!='') b \n" +
                    "ON a.entid = b.entid \n" +
                    "INNER JOIN \n" +
                    "  (SELECT * FROM a_dc_ep_ods.t_eci_company WHERE credit_code IS NOT NULL AND credit_code!='') c \n" +
                    "ON upper(trim(b.creditcode)) = upper(trim(c.credit_code)) \n" +
                    "INNER JOIN  \n" +
                    "(SELECT distinct RECORD_ID,MODIFYID FROM a_dc_ep_ods.zs_t_judicial_ass WHERE MODIFYID IS NOT NULL AND MODIFYID !='') d " +
                    "ON a.MODIFYID = d.MODIFYID");

//        Dataset<Row> update_data = spark.sql("SELECT " +
//                "       a.RECORD_ID as RECORD_ID,\n" +
//                "       c.key_no,\n" +
//                "       c.company_id,\n" +
//                "       c.company_name,\n" +
//                "       a.EXECUTION,\n" +
//                "       a.JUDGMENT_NO,\n" +
//                "       case when a.INAME_LTYPE is not null then a.INAME_LTYPE else a.INAME_ITYPE end, \n" +
//                "       a.INAME_LICENCE,\n" +
//                "       a.ASSIGNEE,\n" +
//                "       a.AEDATE,\n" +
//                "       case when a.ASSIGNEE_LTYPE is not null then a.ASSIGNEE_LTYPE else a.ASSIGNEE_ITYPE end,\n" +
//                "       a.ASSIGNEE_LICENCE,\n" +
//                "       \"\",\n" +
//                "       a.JOBID,\n" +
//                "       \"0\" as isadd  " +
//                "FROM \n" +
//                "  (SELECT * FROM a_dc_ep_ods.zs_t_judicial_ass_alter WHERE ENTID IS NOT NULL AND ENTID !='') a \n" +
//                "INNER JOIN \n" +
//                "  (SELECT * FROM a_dc_ep_ods.zs_t_judicial_ass_del WHERE creditcode IS NOT NULL AND creditcode!='') b \n" +
//                "ON a.entid = b.entid \n" +
//                "INNER JOIN \n" +
//                "  (SELECT * FROM a_dc_ep_ods.t_eci_company WHERE credit_code IS NOT NULL AND credit_code!='') c \n" +
//                "ON upper(trim(b.creditcode)) = upper(trim(c.credit_code)) \n" +
//                "INNER JOIN\n" +
//                "  a_dc_ep_dwi.t_eci_assistance_judicial_partners_change d \n" +
//                "ON (d.company_id = c.company_id and trim(translate(translate(translate(translate(a.JUDGMENT_NO,\"（\",\"\"),\"）\",\"\"),\"(\",\"\"),\")\",\"\"))=trim(translate(translate(translate(translate(d.execution_verdict_num,\"（\",\"\"),\"）\",\"\"),\"(\",\"\"),\")\",\"\"))) \n" +
//                "where b.creditcode is not null and b.creditcode !='' and c.credit_code is not null and c.credit_code!='' ");
            update_data.createOrReplaceTempView("tmp_t_eci_assistance_judicial_partners_change1");
            spark.sql("insert into table a_dc_ep_incr.t_eci_assistance_judicial_partners_change select * from tmp_t_eci_assistance_judicial_partners_change1");

            //获取中数增量数据表中的新增数据
//        Dataset<Row> insert_data = spark.sql("SELECT " +
//                "       a.RECORD_ID as RECORD_ID,\n" +
//                "       md5(if(length(b.creditcode)=18,b.creditcode,c.company_name)) as key_no,\n" +
//                "       md5(if(length(b.creditcode)=18,b.creditcode,c.company_name)) as company_id,\n" +
//                "       c.company_name,\n" +
//                "       a.EXECUTION,\n" +
//                "       a.JUDGMENT_NO,\n" +
//                "       case when a.INAME_LTYPE is not null then a.INAME_LTYPE else a.INAME_ITYPE end, \n" +
//                "       a.INAME_LICENCE,\n" +
//                "       a.ASSIGNEE,\n" +
//                "       a.AEDATE,\n" +
//                "       case when a.ASSIGNEE_LTYPE is not null then a.ASSIGNEE_LTYPE else a.ASSIGNEE_ITYPE end,\n" +
//                "       a.ASSIGNEE_LICENCE,\n" +
//                "       \"\",\n" +
//                "       a.JOBID,\n" +
//                "       \"1\" as isadd  " +
//                "FROM\n" +
//                "  (SELECT * FROM a_dc_ep_ods.zs_t_judicial_ass_alter WHERE ENTID IS NOT NULL AND ENTID !='') a\n" +
//                "INNER JOIN\n" +
//                "  (SELECT * FROM a_dc_ep_ods.zs_t_judicial_ass_del WHERE creditcode IS NOT NULL AND creditcode!='') b \n" +
//                "ON a.entid = b.entid \n" +
//                "left join \n" +
//                "  (SELECT * FROM a_dc_ep_ods.t_eci_company WHERE credit_code IS NOT NULL AND credit_code!='') c \n" +
//                "ON upper(trim(b.creditcode)) = upper(trim(c.credit_code))\n" +
//                "left join \n" +
//                "  (SELECT * FROM tmp_t_eci_assistance_judicial_partners_change1 WHERE RECORD_ID IS NOT NULL AND RECORD_ID!='') d \n" +
//                "ON a.RECORD_ID = d.RECORD_ID \n" +
//                "where d.RECORD_ID is null \n");
//        insert_data.createOrReplaceTempView("tmp_t_eci_assistance_judicial_partners_change2");
//        spark.sql("insert into table a_dc_ep_incr.t_eci_assistance_judicial_partners_change select * from tmp_t_eci_assistance_judicial_partners_change2");
        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_eci_assistance_judicial_partners_change");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }
        spark.stop();
    }
}
